package com.hujf.flink.stock;

import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

/**
 * @Author hujf
 * @Date 2022/4/18 14:48
 * @Description TODO
 */
public class StockSource implements SourceFunction<StockPrice> {

    private boolean isRun = true;

    private String path;
    private InputStream stream;

    public StockSource(String path) {
        this.path = path;
    }

    @Override
    public void run(SourceContext ctx) throws Exception {

        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd HHmmss");

        stream = this.getClass().getClassLoader().getResourceAsStream(path);

        BufferedReader br = new BufferedReader(new InputStreamReader(stream));

        String line;
        boolean isFrist = true;
        long timeDiff = 0;
        long lastEventTs = 0;

        while ( (line = br.readLine()) != null){
            if(isFrist){
                isFrist = false;
            }else{
                String [] itemStrArr = line.split(",");
                LocalDateTime dateTime = LocalDateTime.parse(itemStrArr[1]+" "+itemStrArr[2],formatter);
                long event = Timestamp.valueOf(dateTime).getTime();
                lastEventTs = event;
                StockPrice stockPrice = new StockPrice();
                stockPrice.setSymbol(itemStrArr[0]);
                stockPrice.setPrice(itemStrArr[3]);
                stockPrice.setVolume(Long.parseLong(itemStrArr[4]));
                stockPrice.setTime(itemStrArr[1]);
                stockPrice.setTs(event);

                timeDiff = event - lastEventTs;
                if(timeDiff>0){
                    Thread.sleep(timeDiff);
                }
                ctx.collect(stockPrice);
                lastEventTs = event;
            }


        }
    }

    @Override
    public void cancel() {
        try {
            stream.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        isRun = false;
    }
}
